package com.xhwl.data.mq.producer;

import com.xhwl.common.constant.cent.TopicConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Nonnull;
import java.util.concurrent.Executor;

/**
 * Falcon.cao @Copyright (C), 2022, 兴海物联
 * Date: 2022/7/4 9:10
 * Description: 将组织数据推送到主题
 */
@Slf4j
@Component
public class OrganizationProducer {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Autowired
    private Executor getExecutor;

    public void createPush(String content) {
        getExecutor.execute(() -> {
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TopicConstant.ORGANIZATION_CREATE, content);
            future.addCallback(listener(content));
        });
    }


    public void updatePush(String content) {
        getExecutor.execute(() -> {
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TopicConstant.ORGANIZATION_UPDATE, content);
            future.addCallback(listener(content));
        });
    }


    public void deletePush(String content) {
        getExecutor.execute(() -> {
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TopicConstant.ORGANIZATION_DELETE, content);
            future.addCallback(listener(content));
        });
    }

    private ListenableFutureCallback<SendResult<String, Object>> listener(String content) {
        return new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            //推送成功
            public void onSuccess(SendResult<String, Object> result) {
                log.info("推送成功 - 主题：{}，内容：{}", TopicConstant.ORGANIZATION_CREATE, content);
            }

            @Override
            //推送失败
            public void onFailure(@Nonnull Throwable ex) {
                log.info("推送失败 - 主题：{}，内容：{}，信息：{}", TopicConstant.ORGANIZATION_CREATE, content, ex.getMessage());
            }
        };
    }
}
